package com.htiiot.status.kafka;

import com.alibaba.fastjson.JSONObject;
import com.htiiot.status.write.WriterCache;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.kafka.annotation.KafkaListener;

import java.io.IOException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.atomic.LongAdder;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author 李延念
 * @create 2018-09-26 10:12
 **/
public class Listener {
    private static Logger logger = LogManager.getLogger(Listener.class);
    private static String hostAddress;
    private static ReentrantReadWriteLock lock = new ReentrantReadWriteLock(  );
    public static ReentrantReadWriteLock.ReadLock readLock = lock.readLock();
    public static ReentrantReadWriteLock.WriteLock writeLock = lock.writeLock();
    // 统计数据条数
    public static LongAdder longAdder = new LongAdder();
    static {
        try {
            hostAddress = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
    }

    @KafkaListener(topics = {"dstatus"})
    public void listen(ConsumerRecord<String, String> record) {
        /**
         * {"did":88369,"sts":3000,"tid":3139,"ts":1537765079521}
           {"did":90683,"sts":3000,"tid":3270,"ts":1537765079564}
           {"did":13918,"sts":3000,"tid":60,"ts":1537765079832}
         */
        logger.info("receive value is : " + record.value());
        try {
            JSONObject jsonObject = JSONObject.parseObject( record.value() );
            Object tid = jsonObject.get( "tid" );
            if(tid != null && !tid.equals( "" )){
                readLock.lock();
                try {
                    WriterCache.getWriter(Long.valueOf(tid.toString()),hostAddress).write(record.value()+"\n");
                    longAdder.increment();
                } catch (IOException e) {
                    e.printStackTrace();
                    logger.error( e );
                }finally {
                    readLock.unlock();
                    //logger.info( "receive data size is "+longAdder.sum() );
                }
            }
        }catch (Exception e){
            e.printStackTrace();
            logger.error( e );
        }
    }
}
